package ma.zhiqiang.dataline.core.engine;

import lombok.extern.slf4j.Slf4j;
import ma.zhiqiang.dataline.config.JobConfig;
import ma.zhiqiang.dataline.config.SinkConfig;
import ma.zhiqiang.dataline.config.SourceConfig;
import ma.zhiqiang.dataline.config.TransformConfig;
import ma.zhiqiang.dataline.connector.sink.ConsoleSink;
import ma.zhiqiang.dataline.connector.sink.FileSink;
import ma.zhiqiang.dataline.connector.sink.KafkaSink;
import ma.zhiqiang.dataline.connector.source.FileSource;
import ma.zhiqiang.dataline.connector.transform.ColumnsTransform;
import ma.zhiqiang.dataline.connector.transform.NoneTransform;
import ma.zhiqiang.dataline.connector.transform.ReplaceTransform;

@Slf4j
public class JobFactory {
    JobConfig jobConfig;
    Source source;
    Transform transform;
    Sink sink;

    public JobFactory(JobConfig jobConfig) {
        this.jobConfig = jobConfig;
        init();
    }
    public void init(){
        this.source = initSource(jobConfig.getSource());
        this.transform = initTransForm(jobConfig.getTransform());
        this.sink = initSink(jobConfig.getSink());
    }
    public Pipeline getPipeline(){
        Pipeline pipeline = new Pipeline(source,transform,sink);
        return pipeline;
    }
    private Source initSource(SourceConfig sourceConfig){
        if(null==sourceConfig){throw new RuntimeException("source配置不存在");}
        Source source;
        switch (sourceConfig.getType()){
            default:    // 默认file
                source = new FileSource(sourceConfig.getProperties());
        }
        return source;
    }
    private Transform initTransForm(TransformConfig transformConfig){
        if(null==transformConfig){throw new RuntimeException("transform配置不存在");}
        Transform transform;
        switch (transformConfig.getType()){
            case "replace":
                transform = new ReplaceTransform(transformConfig.getProperties());break;
            case "columns":
                transform = new ColumnsTransform(transformConfig.getProperties());break;
            default:    //默认none
                transform = new NoneTransform();
        }
        return transform;
    }
    public Sink initSink(SinkConfig sinkConfig){
        if(null==sinkConfig){throw new RuntimeException("sink配置不存在");}
        Sink sink;
        switch (sinkConfig.getType()){
            case "file":
                sink = new FileSink(sinkConfig.getProperties()); break;
            case "kafka":
                sink = new KafkaSink(sinkConfig.getProperties()); break;
            default:    //默认console
                sink = new ConsoleSink(sinkConfig.getProperties());
        }
        return sink;
    }
}
